package com.swak.consistency.reliablemsg.handle;

import java.lang.reflect.Method;
import java.util.List;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;

import com.swak.App;
import com.swak.annotation.ReliableMsg;
import com.swak.asm.FieldCache;
import com.swak.asm.FieldCache.ClassMeta;
import com.swak.asm.FieldCache.FieldMeta;
import com.swak.consistency.ConsistencyLauncher;
import com.swak.consistency.core.aspect.ConsistencyProceeding;
import com.swak.consistency.core.execption.ConsistencyException;
import com.swak.consistency.core.handle.ConsistencyHandler;
import com.swak.consistency.core.invoke.Invocation;
import com.swak.consistency.core.invoke.Invoker;
import com.swak.consistency.reliablemsg.config.ReliablemsgConfig;
import com.swak.consistency.reliablemsg.context.Phase;
import com.swak.consistency.reliablemsg.context.ReliableMsgContext;
import com.swak.consistency.reliablemsg.context.ReliableMsgLog;
import com.swak.consistency.reliablemsg.storage.ReliableMsgLogStorage;
import com.swak.consistency.reliablemsg.transport.PrimaryReliableMsgLogTransport;
import com.swak.consistency.reliablemsg.transport.ReliableMsgLogTransport;
import com.swak.define.CurdDefine;
import com.swak.entity.Event;
import com.swak.entity.Event.EventDTO;
import com.swak.utils.StringUtils;
import com.swak.utils.time.DateTimes;

public class ReliablemsgConsistencyHandler implements ConsistencyHandler<ReliableMsgContext> {

	private static Logger logger = LoggerFactory.getLogger(ConsistencyLauncher.class);

	@Autowired
	protected ReliableMsgLogStorage reliableMsgLogStorage;
	@Autowired
	protected Invoker invoker;

	protected ReliablemsgConfig config;
	protected ReliableMsgLogTransport reliableMsgLogTransport;

	public ReliablemsgConsistencyHandler(ReliablemsgConfig config, List<ReliableMsgLogTransport> transports) {
		this.config = config;
		this.reliableMsgLogTransport = new PrimaryReliableMsgLogTransport(transports);
	}

	@Override
	public Object prepare(ReliableMsgContext context) {
		if (logger.isDebugEnabled()) {
			logger.debug("执行可靠消息  第一阶段 (开始)" + "：" + context);
		}

		// 执行结果
		Object result = null;
		Throwable error = null;

		try {

			// 执行业务代码
			result = this.doProceeding(context);

			// 进入确认阶段
			this.doPrepare(context, result);

		} catch (Throwable e) {
			error = e;
		}

		if (logger.isDebugEnabled()) {
			logger.debug("执行可靠消息  第一阶段(结束)" + (error == null ? "成功" : "失败") + "：" + context);
		}

		// 执行期间异常
		if (error != null) {
			return this.throwConsistencyException(error);
		}

		// 返回结果
		return result;
	}

	/**
	 * 执行准备阶段
	 * 
	 * @param context
	 * @return
	 */
	protected Object doProceeding(ReliableMsgContext context) {
		Throwable error = null;
		Object result = null;
		try {
			result = context.getProceeding().proceed();
		} catch (Throwable e) {
			error = e;
		}

		// 抛出异常
		if (error != null) {
			return this.throwConsistencyException(error);
		}

		// 返回结果
		return result;
	}

	/**
	 * 开启可靠消息: 只要执行成功，则表示业务也是成功的
	 * 
	 * @param context
	 */
	@SuppressWarnings("rawtypes")
	protected void doPrepare(ReliableMsgContext context, Object result) {

		// 创建可靠消息
		ReliableMsgLog reliableMsg = new ReliableMsgLog();
		reliableMsg.setMsgId(App.genId());
		reliableMsg.setPhase(Phase.Prepare.getCode());
		reliableMsg.setRetryTimes(0);
		reliableMsg.setVersion(0);

		ConsistencyProceeding proceeding = context.getProceeding();
		Method method = context.getProceeding().getMethod();
		ReliableMsg reliableMessage = method.getAnnotation(ReliableMsg.class);
		reliableMsg.setQueue(reliableMessage.queue());
		reliableMsg.setQueueType(reliableMessage.queueType().name());
		reliableMsg.setCreateTime(DateTimes.now());
		reliableMsg.setUpdateTime(DateTimes.now());
		reliableMsg.setMaxRts(reliableMessage.retryTimes() <= 0 //
				? config.getRetryTimes() // 默认的重试次数
				: reliableMessage.retryTimes());// 自定义的重试次数

		// 是否存储消息 reliableMessage.retryTimes() == 0 不存储消息， 不需要重試
		context.setStoreMsg(reliableMessage.retryTimes() != 0);

		// 默认将返回结果作为消息结果
		Object message = result;

		// 添加事务第二阶段方法
		Invocation confirmInvocation = null;
		String confirmMethodName = reliableMessage.confirm();
		if (StringUtils.isNotBlank(confirmMethodName)) {
			confirmInvocation = invoker.createInvocation(proceeding, confirmMethodName);
			message = this.invokeInvocation(context, confirmInvocation);
		}

		// message 是否实现了Event
		if (message != null && message instanceof Event) {
			message = ((Event) message).build(reliableMessage.event());
		}

		// 约定如果没有返回值则不会发送可靠消息
		if (message != null) {

			/*
			 * 对消息的封装： 如果设置为 -128（默认），不会封装消息
			 */
			Object messageWrap = !(message instanceof EventDTO || reliableMessage.event() == -128)
					? EventDTO.of().setData(message).setType(reliableMessage.event())
					: message;

			/*
			 * 尝试获取ID
			 */
			if (messageWrap instanceof EventDTO // 转为 EventDTO
					&& ((EventDTO) messageWrap).getId() == null // 默认没有ID值
					&& FieldCache.exists(result.getClass())) { // 如果是缓存的类型
				try {
					ClassMeta classMeta = FieldCache.get(result.getClass());
					FieldMeta fieldMeta = classMeta.getField(CurdDefine.IdProperty);
					if (fieldMeta != null) {
						Object value = fieldMeta.getFieldValue(result);
						((EventDTO) messageWrap).setId(value);
					}
				} catch (Exception e) {
					// ig
				}
			}

			// 执行确认
			reliableMsg.setMessage(messageWrap);

			// 保存事务 -- 事务不成功直接抛出异常
			if (context.isStoreMsg()) {
				this.reliableMsgLogStorage.create(reliableMsg);
			}

			// 设置当前事务上下文
			context.update(reliableMsg, Phase.Prepare);
		}
	}

	/**
	 * 执行 TransInvocation
	 * 
	 * @param invocation
	 * @throws Exception
	 */
	protected Object invokeInvocation(ReliableMsgContext context, Invocation invocation) {
		// 没有参与者
		if (invocation == null) {
			return null;
		}
		try {
			return invoker.invokeInvocation(invocation);
		} catch (Throwable e) {
			logger.error("构建消息异常：" + invocation + "; 事务属性：" + context, e);
		}
		throw new ConsistencyException("构建消息异常：" + invocation + "; 事务属性：" + context);
	}

	@Override
	public Object complete(ReliableMsgContext context) {

		// 执行结果
		Object result = null;
		Throwable error = null;

		try {
			// 执行准备阶段
			result = context.getProceeding().proceed();
		} catch (Throwable e) {
			error = e;
		}

		// 执行期间异常
		if (error != null) {
			this.quietExecute(() -> {
				this.doCancel(context);
			});
			return this.throwConsistencyException(error);
		}

		// 确认提交
		this.quietExecute(() -> {
			this.doConfirm(context);
		});

		// 返回结果
		return result;
	}

	@Override
	public Object transmit(ReliableMsgContext context) {
		throw new ConsistencyException("不支持transmit");
	}

	@Override
	public void retry(ReliableMsgContext context) {
		if (context.getPhase() == Phase.Prepare) {
			context.getTrans().setPhase(Phase.Confirm.getCode());
			this.reliableMsgLogTransport.send(context.getTrans());
		}
		this.reliableMsgLogStorage.remove(context.getTrans());
	}

	/**
	 * 确认事务
	 * 
	 * @param context
	 */
	protected void doConfirm(ReliableMsgContext context) {

		if (logger.isDebugEnabled()) {
			logger.debug("执行可靠消息  第二阶段(开始)" + "：" + context);
		}

		// 不需要执行可靠消息
		if (context.getMsgId() == null) {
			return;
		}

		// 处理确认 -- 如果不存储消息，不需要重试
		Throwable error = null;
		try {
			ReliableMsgLog reliableMsg = context.isStoreMsg() ? this.reliableMsgLogStorage.get(context.getMsgId())
					: context.getTrans();
			reliableMsg.setPhase(Phase.Confirm.getCode());
			this.reliableMsgLogTransport.send(reliableMsg);
			if (context.isStoreMsg()) {
				this.reliableMsgLogStorage.updatePhase(reliableMsg);
			}
		} catch (Throwable e) {
			error = e;
			// ignore
		}

		if (logger.isDebugEnabled() && error == null) {
			logger.debug("执行可靠消息  第二阶段(结束) 成功 ：" + context);
		} else if (logger.isDebugEnabled()) {
			logger.error("执行可靠消息  第二阶段(结束) 失败，后续开启异步重试，" + context, error);
		}
	}

	/**
	 * 确认事务
	 * 
	 * @param context
	 */
	protected void doCancel(ReliableMsgContext context) {

		// 保存一定有消息ID
		if (context.getMsgId() == null) {
			return;
		}

		// 处理失败阶段
		Throwable error = null;
		try {
			ReliableMsgLog reliableMsg = context.isStoreMsg() ? this.reliableMsgLogStorage.get(context.getMsgId())
					: context.getTrans();
			reliableMsg.setPhase(Phase.Cancel.getCode());
			if (context.isStoreMsg()) {
				this.reliableMsgLogStorage.updatePhase(reliableMsg);
			}
		} catch (Throwable e) {
			error = e;
			// ignore
		}

		if (logger.isDebugEnabled() && error == null) {
			logger.debug("取消阶段执行成功 ：" + context);
		} else if (logger.isDebugEnabled()) {
			logger.error("取消阶段异常，部分参与者取消阶段未执行成功，后续开启异步重试，" + context);
		}
	}

	// 不抛出异常
	private void quietExecute(Runnable run) {
		try {
			run.run();
		} catch (Throwable e) {
			// ig
		}
	}
}
